package com.yaxin.mq.callback;

import com.yaxin.mq.entity.CustomCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * RabbitMQ回调处理，支持确认失败 & 路由失败的动态重试
 */
@Slf4j
@Component
public class RabbitCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    private static final int MAX_RETRY_COUNT = 3;

    // 内存缓存确认失败的重试次数（可替换为Redis/DB）
    private final Map<String, Integer> retryCountCache = new ConcurrentHashMap<>();

    private RabbitTemplate rabbitTemplate;

    public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    /**
     * 发送确认回调（消息到达Exchange）
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData.getId();
        if (ack) {
            log.info("【确认成功】消息ID={}，消息已成功到达交换机", id);
            retryCountCache.remove(id);
        } else {
            log.error("【确认失败】消息ID={}，原因：{}", id, cause);
            handleFailedMessage(correlationData);
        }
    }

    /**
     * 路由失败回调（Exchange无队列匹配）
     */
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        Message message = returnedMessage.getMessage();
        String body = new String(message.getBody(), StandardCharsets.UTF_8);

        log.error("【路由失败】ReplyCode: {}, ReplyText: {}, Exchange: {}, RoutingKey: {}, 消息内容: {}",
            returnedMessage.getReplyCode(),
            returnedMessage.getReplyText(),
            returnedMessage.getExchange(),
            returnedMessage.getRoutingKey(),
            body);

        retrySendByReturnedMessage(returnedMessage);
    }

    /**
     * 确认失败重试
     */
    private void handleFailedMessage(CorrelationData correlationData) {
        if (!(correlationData instanceof CustomCorrelationData)) {
            log.error("【确认失败异常】correlationData类型错误，期望CustomCorrelationData，实际：{}",
                correlationData.getClass().getSimpleName());
            return;
        }

        CustomCorrelationData customData = (CustomCorrelationData) correlationData;
        String id = customData.getId();

        int retryCount = retryCountCache.getOrDefault(id, 0);
        if (retryCount < MAX_RETRY_COUNT) {
            retryCount++;
            retryCountCache.put(id, retryCount);

            log.warn("【确认失败重试】消息ID={}，Exchange={}，RoutingKey={}，当前重试次数={}",
                id, customData.getExchange(), customData.getRoutingKey(), retryCount);

            rabbitTemplate.convertAndSend(
                customData.getExchange(),
                customData.getRoutingKey(),
                customData.getMessageBody(),
                customData
            );
        } else {
            log.error("【确认失败放弃】消息ID={}，超过最大重试次数{}，记录异常或通知人工处理。", id, MAX_RETRY_COUNT);
            retryCountCache.remove(id);
            // TODO：可扩展通知运维、报警、写数据库
        }
    }

    /**
     * 路由失败重试
     */
    private void retrySendByReturnedMessage(ReturnedMessage returnedMessage) {
        Message message = returnedMessage.getMessage();
        String body = new String(message.getBody(), StandardCharsets.UTF_8);

        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        Integer retryCount = (Integer) headers.getOrDefault("x-retry-count", 0);

        if (retryCount < MAX_RETRY_COUNT) {
            retryCount++;
            headers.put("x-retry-count", retryCount);

            log.warn("【路由失败重试】Exchange={}，RoutingKey={}，当前重试次数={}，消息内容：{}",
                returnedMessage.getExchange(),
                returnedMessage.getRoutingKey(),
                retryCount,
                body);

            rabbitTemplate.send(returnedMessage.getExchange(), returnedMessage.getRoutingKey(), message);
        } else {
            log.error("【路由失败放弃】消息内容：{}，已重试{}次，记录异常或转入死信队列。", body, MAX_RETRY_COUNT);
            // TODO：可扩展通知运维、写入异常库、存死信
        }
    }
}
